package com.wande.dataplatform.filecollection.executor;

import cn.hutool.core.io.FileUtil;
import cn.hutool.core.util.IdUtil;
import cn.hutool.json.JSONUtil;
import com.wande.common.core.utils.StringUtils;
import com.wande.dataplatform.filecollection.common.constant.FileCollectionConstants;
import com.wande.dataplatform.filecollection.common.enums.FileCollectionErrorCode;
import com.wande.dataplatform.filecollection.common.exception.FileCollectionException;
import com.wande.dataplatform.filecollection.common.utils.MinioUtils;
import com.wande.dataplatform.filecollection.config.FileCollectionProperties;
import com.wande.dataplatform.filecollection.domain.CollectionError;
import com.wande.dataplatform.filecollection.domain.CollectionRule;
import com.wande.dataplatform.filecollection.domain.CollectionTask;
import com.wande.dataplatform.filecollection.domain.FileRecord;
import com.wande.dataplatform.filecollection.domain.MappingRule;
import com.wande.dataplatform.filecollection.domain.dto.LoadResult;
import com.wande.dataplatform.filecollection.domain.dto.ParseResult;
import com.wande.dataplatform.filecollection.parser.ParserConfig;
import com.wande.dataplatform.filecollection.mapper.CollectionErrorMapper;
import com.wande.dataplatform.filecollection.mapper.CollectionTaskMapper;
import com.wande.dataplatform.filecollection.mapper.FileRecordMapper;
import com.wande.dataplatform.filecollection.mapper.MappingRuleMapper;
import com.wande.dataplatform.filecollection.service.IDataLoadService;
import com.wande.dataplatform.filecollection.service.IFileParsingService;
import com.wande.dataplatform.filecollection.source.IFileSource;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Component;

import java.util.concurrent.Executor;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

import java.io.File;
import java.io.FileInputStream;
import java.io.InputStream;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 采集任务执行器
 *
 * @author wande
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class CollectionTaskExecutor {

    private final CollectionTaskMapper collectionTaskMapper;
    private final FileRecordMapper fileRecordMapper;
    private final CollectionErrorMapper collectionErrorMapper;
    private final MappingRuleMapper mappingRuleMapper;
    private final IFileParsingService fileParsingService;
    private final IDataLoadService dataLoadService;
    private final MinioUtils minioUtils;
    private final FileCollectionProperties properties;
    private final Map<String, IFileSource> fileSourceMap;

    /**
     * 运行中的任务
     */
    private final Map<Long, Boolean> runningTasks = new ConcurrentHashMap<>();

    /**
     * 性能监控器（可选）
     */
    @Autowired(required = false)
    private com.wande.dataplatform.filecollection.performance.PerformanceMonitor performanceMonitor;

    /**
     * 文件处理线程池（可选）
     */
    @Autowired(required = false)
    @Qualifier("fileProcessingExecutor")
    private Executor fileProcessingExecutor;

    /**
     * 告警服务（可选）
     */
    @Autowired(required = false)
    private com.wande.dataplatform.filecollection.alert.AlertService alertService;

    /**
     * 执行采集任务
     *
     * @param rule 采集规则
     * @return 任务ID
     */
    public Long executeTask(CollectionRule rule) {
        // 创建任务记录
        CollectionTask task = createTask(rule);
        
        // 开始性能监控
        if (performanceMonitor != null) {
            performanceMonitor.startTask(task.getId());
        }
        
        try {
            // 标记任务为运行中
            runningTasks.put(task.getId(), true);
            
            // 更新任务状态为运行中
            updateTaskStatus(task.getId(), FileCollectionConstants.TASK_STATUS_RUNNING, null);
            
            // 获取映射规则
            MappingRule mappingRule = mappingRuleMapper.selectById(rule.getMappingRuleId());
            if (mappingRule == null) {
                throw new FileCollectionException(FileCollectionErrorCode.MAPPING_RULE_NOT_FOUND,
                    "映射规则不存在: " + rule.getMappingRuleId());
            }
            
            // 获取文件列表
            List<File> files = fetchFiles(rule);
            task.setFileCount(files.size());
            collectionTaskMapper.updateById(task);
            
            log.info("开始执行采集任务: taskId={}, ruleId={}, fileCount={}", 
                task.getId(), rule.getId(), files.size());
            
            // 判断是否使用并行处理
            boolean useParallel = fileProcessingExecutor != null && files.size() > 1;
            
            if (useParallel) {
                // 并行处理文件
                processFilesInParallel(task, rule, mappingRule, files);
            } else {
                // 串行处理文件
                for (File file : files) {
                    if (!isTaskRunning(task.getId())) {
                        log.info("任务已停止: taskId={}", task.getId());
                        break;
                    }
                    
                    processFile(task, rule, mappingRule, file);
                }
            }
            
            // 更新任务状态为成功
            updateTaskStatus(task.getId(), FileCollectionConstants.TASK_STATUS_SUCCESS, null);
            
            // 结束性能监控
            if (performanceMonitor != null) {
                performanceMonitor.endTask(task.getId(), true);
            }
            
            log.info("采集任务执行成功: taskId={}", task.getId());
            
        } catch (Exception e) {
            log.error("采集任务执行失败: taskId={}", task.getId(), e);
            updateTaskStatus(task.getId(), FileCollectionConstants.TASK_STATUS_FAILED, e.getMessage());
            
            // 结束性能监控
            if (performanceMonitor != null) {
                performanceMonitor.endTask(task.getId(), false);
            }
            
            // 发送告警
            if (alertService != null) {
                alertService.sendTaskFailureAlert(task.getId(), rule.getRuleName(), e.getMessage());
            }
            
            // 重试机制
            int retryTimes = properties.getLoader().getRetryTimes();
            if (task.getErrorMessage() != null && !task.getErrorMessage().contains("重试")) {
                for (int i = 1; i <= retryTimes; i++) {
                    try {
                        log.info("开始第{}次重试: taskId={}", i, task.getId());
                        Thread.sleep(5000 * i); // 递增延迟
                        executeTask(rule);
                        return task.getId();
                    } catch (Exception ex) {
                        log.error("第{}次重试失败: taskId={}", i, task.getId(), ex);
                    }
                }
            }
        } finally {
            // 移除运行标记
            runningTasks.remove(task.getId());
        }
        
        return task.getId();
    }

    /**
     * 并行处理文件
     */
    private void processFilesInParallel(CollectionTask task, CollectionRule rule, 
                                       MappingRule mappingRule, List<File> files) {
        List<CompletableFuture<Void>> futures = new ArrayList<>();
        
        for (File file : files) {
            if (!isTaskRunning(task.getId())) {
                log.info("任务已停止: taskId={}", task.getId());
                break;
            }
            
            CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
                processFile(task, rule, mappingRule, file);
            }, fileProcessingExecutor);
            
            futures.add(future);
        }
        
        // 等待所有文件处理完成
        try {
            CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).get();
            log.info("并行处理完成: taskId={}, fileCount={}", task.getId(), files.size());
        } catch (InterruptedException | ExecutionException e) {
            log.error("并行处理失败: taskId={}", task.getId(), e);
            throw new FileCollectionException(FileCollectionErrorCode.FILE_PROCESS_ERROR,
                "并行处理失败: " + e.getMessage());
        }
    }

    /**
     * 停止任务
     *
     * @param taskId 任务ID
     */
    public void stopTask(Long taskId) {
        runningTasks.put(taskId, false);
        log.info("停止任务: taskId={}", taskId);
    }

    /**
     * 检查任务是否运行中
     */
    private boolean isTaskRunning(Long taskId) {
        return Boolean.TRUE.equals(runningTasks.get(taskId));
    }

    /**
     * 创建任务记录
     */
    private CollectionTask createTask(CollectionRule rule) {
        CollectionTask task = new CollectionTask();
        task.setRuleId(rule.getId());
        task.setTaskCode(IdUtil.fastSimpleUUID());
        task.setStatus(FileCollectionConstants.TASK_STATUS_PENDING);
        task.setStartTime(LocalDateTime.now());
        task.setFileCount(0);
        task.setTotalRecords(0L);
        task.setSuccessRecords(0L);
        task.setFailedRecords(0L);
        
        collectionTaskMapper.insert(task);
        return task;
    }

    /**
     * 更新任务状态
     */
    private void updateTaskStatus(Long taskId, String status, String errorMessage) {
        CollectionTask task = collectionTaskMapper.selectById(taskId);
        if (task != null) {
            task.setStatus(status);
            if (FileCollectionConstants.TASK_STATUS_SUCCESS.equals(status) || 
                FileCollectionConstants.TASK_STATUS_FAILED.equals(status)) {
                task.setEndTime(LocalDateTime.now());
            }
            if (StringUtils.isNotEmpty(errorMessage)) {
                task.setErrorMessage(errorMessage);
            }
            collectionTaskMapper.updateById(task);
        }
    }

    /**
     * 获取文件列表
     */
    private List<File> fetchFiles(CollectionRule rule) {
        try {
            IFileSource fileSource = fileSourceMap.get(rule.getSourceType().toLowerCase() + "FileSource");
            if (fileSource == null) {
                throw new FileCollectionException(FileCollectionErrorCode.FILE_SOURCE_NOT_FOUND,
                    "不支持的文件来源类型: " + rule.getSourceType());
            }
            
            String sourceConfigJson = rule.getSourceConfig();
            List<String> filePaths = fileSource.listFiles(sourceConfigJson);
            
            // 将文件路径转换为File对象
            List<File> files = new ArrayList<>();
            for (String filePath : filePaths) {
                files.add(new File(filePath));
            }
            return files;
            
        } catch (Exception e) {
            log.error("获取文件列表失败", e);
            throw new FileCollectionException(FileCollectionErrorCode.FILE_FETCH_ERROR,
                "获取文件列表失败: " + e.getMessage());
        }
    }

    /**
     * 处理单个文件
     */
    private void processFile(CollectionTask task, CollectionRule rule, MappingRule mappingRule, File file) {
        FileRecord fileRecord = null;
        long startTime = System.currentTimeMillis();
        
        try {
            // 创建文件记录
            fileRecord = createFileRecord(task.getId(), file, rule.getFileType());
            
            // 上传文件到MinIO
            String ossPath = uploadFileToMinio(file);
            fileRecord.setOssPath(ossPath);
            fileRecordMapper.updateById(fileRecord);
            
            // 更新文件记录状态为解析中
            updateFileRecordStatus(fileRecord.getId(), FileCollectionConstants.FILE_STATUS_PARSING, null);
            
            // 解析文件
            long parseStartTime = System.currentTimeMillis();
            ParseResult parseResult;
            try (InputStream inputStream = new FileInputStream(file)) {
                ParserConfig parserConfig = ParserConfig.builder().build();
                parseResult = fileParsingService.parseFile(inputStream, rule.getFileType(), parserConfig);
            }
            long parseTime = System.currentTimeMillis() - parseStartTime;
            
            if (!parseResult.getSuccess()) {
                throw new FileCollectionException(FileCollectionErrorCode.FILE_PARSE_ERROR,
                    "文件解析失败: " + parseResult.getErrorMessage());
            }
            
            fileRecord.setParsedRecords((long) parseResult.getTotalRecords());
            fileRecordMapper.updateById(fileRecord);
            
            // 记录解析性能
            if (performanceMonitor != null) {
                performanceMonitor.recordParsePerformance(task.getId(), parseTime, parseResult.getTotalRecords());
            }
            
            // 加载数据
            long loadStartTime = System.currentTimeMillis();
            LoadResult loadResult = dataLoadService.loadDataBatch(
                parseResult.getData().iterator(), rule, mappingRule);
            long loadTime = System.currentTimeMillis() - loadStartTime;
            
            // 记录加载性能
            if (performanceMonitor != null) {
                performanceMonitor.recordLoadPerformance(task.getId(), loadTime, 
                    loadResult.getInsertedRecords() + loadResult.getUpdatedRecords());
            }
            
            // 更新文件记录
            fileRecord.setLoadedRecords(loadResult.getInsertedRecords() + loadResult.getUpdatedRecords());
            
            if (loadResult.getSuccess()) {
                updateFileRecordStatus(fileRecord.getId(), FileCollectionConstants.FILE_STATUS_SUCCESS, null);
            } else {
                updateFileRecordStatus(fileRecord.getId(), FileCollectionConstants.FILE_STATUS_FAILED, 
                    loadResult.getErrorMessage());
            }
            
            // 更新任务统计
            updateTaskStatistics(task.getId(), loadResult);
            
            // 保存错误记录
            saveErrorRecords(task.getId(), fileRecord.getId(), loadResult);
            
            // 记录文件处理性能
            long totalTime = System.currentTimeMillis() - startTime;
            if (performanceMonitor != null) {
                performanceMonitor.recordFileProcessed(task.getId(), file.length(), 
                    fileRecord.getLoadedRecords(), totalTime);
            }
            
            log.info("文件处理完成: fileId={}, fileName={}, parsed={}, loaded={}, time={}ms", 
                fileRecord.getId(), file.getName(), fileRecord.getParsedRecords(), 
                fileRecord.getLoadedRecords(), totalTime);
            
        } catch (Exception e) {
            log.error("文件处理失败: fileName={}", file.getName(), e);
            
            if (fileRecord != null) {
                updateFileRecordStatus(fileRecord.getId(), FileCollectionConstants.FILE_STATUS_FAILED, 
                    e.getMessage());
                
                // 更新任务失败记录数
                CollectionTask currentTask = collectionTaskMapper.selectById(task.getId());
                currentTask.setFailedRecords(currentTask.getFailedRecords() + 1);
                collectionTaskMapper.updateById(currentTask);
            }
        } finally {
            // 清理临时文件
            if (file.exists()) {
                FileUtil.del(file);
            }
        }
    }

    /**
     * 创建文件记录
     */
    private FileRecord createFileRecord(Long taskId, File file, String fileType) {
        FileRecord record = new FileRecord();
        record.setTaskId(taskId);
        record.setFileName(file.getName());
        record.setFilePath(file.getAbsolutePath());
        record.setFileType(fileType);
        record.setFileSize(file.length());
        record.setFileMd5(cn.hutool.crypto.SecureUtil.md5(file));
        record.setStatus(FileCollectionConstants.FILE_STATUS_PENDING);
        record.setParsedRecords(0L);
        record.setLoadedRecords(0L);
        
        fileRecordMapper.insert(record);
        return record;
    }

    /**
     * 更新文件记录状态
     */
    private void updateFileRecordStatus(Long fileId, String status, String errorMessage) {
        FileRecord record = fileRecordMapper.selectById(fileId);
        if (record != null) {
            record.setStatus(status);
            if (StringUtils.isNotEmpty(errorMessage)) {
                record.setErrorMessage(errorMessage);
            }
            fileRecordMapper.updateById(record);
        }
    }

    /**
     * 上传文件到MinIO
     */
    private String uploadFileToMinio(File file) {
        try {
            String objectName = "file-collection/" + LocalDateTime.now().toLocalDate() + "/" + file.getName();
            try (InputStream inputStream = new FileInputStream(file)) {
                minioUtils.uploadFile(inputStream, objectName, file.getName(), false);
            }
            return objectName;
        } catch (Exception e) {
            log.error("上传文件到MinIO失败", e);
            throw new FileCollectionException(FileCollectionErrorCode.FILE_UPLOAD_ERROR,
                "上传文件失败: " + e.getMessage());
        }
    }

    /**
     * 更新任务统计
     */
    private void updateTaskStatistics(Long taskId, LoadResult loadResult) {
        CollectionTask task = collectionTaskMapper.selectById(taskId);
        if (task != null) {
            task.setTotalRecords(task.getTotalRecords() + loadResult.getTotalRecords());
            task.setSuccessRecords(task.getSuccessRecords() + 
                loadResult.getInsertedRecords() + loadResult.getUpdatedRecords());
            task.setFailedRecords(task.getFailedRecords() + loadResult.getFailedRecords());
            collectionTaskMapper.updateById(task);
        }
    }

    /**
     * 保存错误记录
     */
    private void saveErrorRecords(Long taskId, Long fileId, LoadResult loadResult) {
        if (loadResult.getErrors() != null && !loadResult.getErrors().isEmpty()) {
            List<CollectionError> errors = new ArrayList<>();
            
            for (LoadResult.LoadError error : loadResult.getErrors()) {
                CollectionError collectionError = new CollectionError();
                collectionError.setTaskId(taskId);
                collectionError.setFileId(fileId);
                collectionError.setRecordIndex(error.getRecordIndex());
                collectionError.setErrorType(error.getErrorType());
                collectionError.setErrorMessage(error.getErrorMessage());
                collectionError.setRecordData(error.getRecordData());
                errors.add(collectionError);
            }
            
            // 批量插入错误记录
            for (CollectionError error : errors) {
                collectionErrorMapper.insert(error);
            }
            
            log.info("保存错误记录: taskId={}, fileId={}, errorCount={}", 
                taskId, fileId, errors.size());
        }
    }
}
